//! A log consists of a number of blocks.
//! A block consists of a number of records and an optional trailer (filler).
//! A record is a bytestring: [checksum: uint32, length: uint16, type: uint8, data: [u8]]
//! checksum is the crc32 sum of type and data; type is one of RecordType::{Full/First/Middle/Last}

use crate::crc;
use crate::error::{err, Result, StatusCode};

use std::io::{Read, Write};

use integer_encoding::FixedInt;
use integer_encoding::FixedIntWriter;

/// 表示数据块的大小为 32KB（32×1024 字节）。
/// 这通常是日志文件或存储系统中数据块的固定大小，用于将数据分块存储和处理。这样的设计有助于便于内存管理、I/O 操作优化，以及数据校验和验证（按块计算校验和）
const BLOCK_SIZE: usize = 32 * 1024;
/// 表示每条日志记录头部的大小为 7 字节（4+2+1）。
/// 头部通常包含元数据，根据常见设计，这 7 字节可能分配为：
/// 4 字节：记录长度（表示后续数据的字节数）
/// 2 字节：某种标志或类型（区分记录类型，如正常记录、删除标记等）
/// 1 字节：校验和标志或其他控制信息（如是否启用校验和）
const HEADER_SIZE: usize = 4 + 2 + 1;

/// 这个 RecordType 枚举定义了日志记录的不同类型，通常用于处理跨数据块的日志记录（当一条日志记录过大，无法放入单个块时）。
#[derive(Clone, Copy)]
pub enum RecordType {
    /// 表示这是一条完整的日志记录，整个记录内容包含在当前数据块中，不需要需要跨块存储。
    Full = 1,
    /// 表示这是一条大记录的第一部分。当一条记录长度超过单个块的剩余空间时，会先存储该记录的起始部分，并标记为 First。
    First = 2,
    /// 表示这是一条大记录的中间部分。当记录长度超过两个块时，中间被分割的部分会标记为 Middle。
    Middle = 3,
    /// 表示这是一条大记录的最后一部分。用于标识跨块记录的结束，与前面的 First 和 Middle 部分组合可还原完整记录
    Last = 4,
}

/// 这个 LogWriter 结构体用于向底层写入流（如文件）按固定格式写入日志记录，通常用于实现数据库中的 WAL（Write-Ahead Log，预写日志）功能，确保数据写入的可靠性和可恢复性。
pub struct LogWriter<W: Write> {
    /// 日志的底层写入目标，是一个实现了 Write 特性的类型（例如文件、网络流或内存缓冲区）。
    /// LogWriter 通过 dst 将格式化后的日志记录持久化到底层存储（如磁盘文件）。
    /// BufWriter::new(PosixDiskEnv)
    dst: W,
    /// 当前数据块中的偏移量，用于跟踪在当前日志块中已写入的字节数。
    /// 日志通常按固定大小的 “块”（block_size）组织，这个字段记录当前块已用空间，用于判断是否需要新建块。
    current_block_offset: usize,
    /// 日志块的固定大小（例如 32KB），是日志数据的基本组织单位。
    /// 当 current_block_offset 加上新记录的长度超过 block_size 时，LogWriter 会自动切换到新块（填充当前块剩余空间为无效数据），保证每条记录的完整性（不会跨块存储头部信息）。
    block_size: usize,
}

impl<W: Write> LogWriter<W> {
    pub fn new(writer: W) -> LogWriter<W> {
        LogWriter {
            dst: writer,
            current_block_offset: 0,
            block_size: BLOCK_SIZE,
        }
    }

    /// new_with_off opens a writer starting at some offset of an existing log file. The file must
    /// have the default block size.
    pub fn new_with_off(writer: W, off: usize) -> LogWriter<W> {
        let mut w = LogWriter::new(writer);
        w.current_block_offset = off % BLOCK_SIZE;
        w
    }

    /// 添加记录到日志文件中, 这里主要计算大小和类型, 在嵌套的方法里会往日志文件里写入
    ///
    /// 这段这段代码定义了一个名为 add_record 的方法，用于将 记录（record）也就是插入进跳表的数据 添加到日志文件中。
    ///
    /// 它的核心功能是处理记录的分片存储，确保大记录能被拆分成适合块大小的片段写入，并在过程中计算大小和类型。
    pub fn add_record(&mut self, mut record: &[u8]) -> Result<usize> {
        let mut first_frag = true;
        let mut result = Ok(0);
        // 使用 while 循环持续处理记录，直到记录被完全写入（record.is_empty()）或出现错误（result.is_ok() 为假）。
        while result.is_ok() && !record.is_empty() {
            // 检查当前块的剩余空间（space_left），如果剩余空间不足以容纳记录头部（HEADER_SIZE），则用空字节填充剩余空间，并切换到新块（current_block_offset = 0）。
            assert!(self.block_size > HEADER_SIZE);

            let space_left = self.block_size - self.current_block_offset;

            // Fill up block; go to next block.
            if space_left < HEADER_SIZE {
                self.dst.write_all(&vec![0, 0, 0, 0, 0, 0][0..space_left])?;
                self.current_block_offset = 0;
            }
            // 计算当前块可用于存储数据的空间（avail_for_data），根据记录剩余长度与可用空间的比较，确定当前分片的长度（data_frag_len）。
            let avail_for_data = self.block_size - self.current_block_offset - HEADER_SIZE;

            let data_frag_len = if record.len() < avail_for_data {
                record.len()
            } else {
                avail_for_data
            };

            let recordtype;
            // 根据分片的位置（首片、中间片、末片）和是否为完整记录，设置记录类型（RecordType）：
            // Full：记录完整且无需分片
            // First：记录的第一片（需要后续分片）
            // Last：记录的最后一片
            // Middle：记录的中间分片
            if first_frag && data_frag_len == record.len() {
                recordtype = RecordType::Full;
            } else if first_frag {
                recordtype = RecordType::First;
            } else if data_frag_len == record.len() {
                recordtype = RecordType::Last;
            } else {
                recordtype = RecordType::Middle;
            }

            // 调用 emit_record 方法将当前分片写入日志文件（实际写入操作在此方法中完成）。
            result = self.emit_record(recordtype, record, data_frag_len);
            // 更新接下来循环要拿的record, 并标记已处理首片（first_frag = false）。
            record = &record[data_frag_len..];
            first_frag = false;
        }
        result
    }

    /// 这个 emit_record 方法是 LogWriter 结构体的核心日志写入逻辑，用于将一条日志记录按固定格式编码并写入到底层存储。同时，它会处理数据的校验和计算，确保日志记录的完整性。
    fn emit_record(&mut self, t: RecordType, data: &[u8], len: usize) -> Result<usize> {
        assert!(len < 256 * 256);

        let mut digest = crc::digest();
        digest.update(&[t as u8]);  // 加入记录类型（转为字节）
        digest.update(&data[0..len]); // 加入原始数据

        let chksum = mask_crc(digest.finalize());  // 生成并处理 CRC 校验和

        let mut s = 0;
        // 1. 写入校验和（4字节，通过 encode_fixed_vec 转为字节流）
        s += self.dst.write(&chksum.encode_fixed_vec())?;
        // 2. 写入数据长度（2字节，u16 类型）
        s += self.dst.write_fixedint(len as u16)?;
        // 3. 写入记录类型（1字节，RecordType 转为 u8）
        s += self.dst.write(&[t as u8])?;
        // 4. 写入原始数据（len 字节）
        s += self.dst.write(&data[0..len])?;
        // 5. 当前数据块中的偏移量，用于跟踪在当前日志块中已写入的字节数
        self.current_block_offset += s;
        Ok(s)
    }

    /// 强制刷盘，确保元数据持久化
    pub fn flush(&mut self) -> Result<()> {
        self.dst.flush()?;
        Ok(())
    }
}

/// 这个 LogReader 结构体用于从一个读取源（R: Read）读取日志数据，通常用于处理结构化的日志文件（类似数据库中的操作日志或 WAL - Write - Ahead Log）。
pub struct LogReader<R: Read> {
    // TODO: Wrap src in a buffer to enhance read performance.
    /// 日志数据的读取源，它实现了 Read 特性，可以是文件、网络流等任何可读取的对象。
    src: R,
    /// 当前在块中的偏移量，用于跟踪在当前数据块中的读取位置。
    blk_off: usize,
    /// 日志文件中数据块的大小，日志可能被分割成固定大小的块进行存储和处理。
    blocksize: usize,
    /// 一个固定大小的字节数组缓冲区，用于临时存储日志记录的头部信息（7 字节可能包含记录长度、类型等元数据）。
    head_scratch: [u8; 7],
    /// 一个布尔值，表示是否启用校验和验证。当为 true 时，读取日志时会验证数据的校验和，以确保数据的完整性，防止数据损坏。
    checksums: bool,
}

impl<R: Read> LogReader<R> {
    pub fn new(src: R, chksum: bool) -> LogReader<R> {
        LogReader {
            src,
            blk_off: 0,
            blocksize: BLOCK_SIZE,
            checksums: chksum,
            head_scratch: [0; 7],
        }
    }

    /// EOF is signalled by Ok(0)
    /// 用于从日志文件中读取完整的日志记录（可能跨多个数据块），并将结果存入 dst 缓冲区。
    pub fn read(&mut self, dst: &mut Vec<u8>) -> Result<usize> {
        // 存储校验和
        let mut checksum: u32;
        // 记录长度
        let mut length: u16;
        // 记录类型
        let mut typ: u8;
        // 目标缓冲区的偏移量
        let mut dst_offset: usize = 0;
        // 清空目标缓冲区 dst，准备接收新数据。
        dst.clear();

        loop {
            //
            if self.blocksize - self.blk_off < HEADER_SIZE {
                // 若当前块剩余空间不足存放头部（HEADER_SIZE=7字节），则跳过该块的剩余部分
                // skip to next block
                self.src.read_exact(&mut self.head_scratch[0..self.blocksize - self.blk_off])?;
                // 切换到下一个块
                self.blk_off = 0;
            }
            // 从文件或者流中读取出数据后存放进 head_scratch , 并把数据长度给到 bytes_read
            let mut bytes_read = self.src.read(&mut self.head_scratch)?;

            // EOF  若读取到0字节，说明已到文件末尾（EOF）
            if bytes_read == 0 {
                return Ok(0);
            }

            self.blk_off += bytes_read;
            // 前4字节：校验和
            checksum = u32::decode_fixed(&self.head_scratch[0..4]);
            // 中间2字节：记录数据长度
            length = u16::decode_fixed(&self.head_scratch[4..6]);
            // 最后1字节：记录类型（RecordType）
            typ = self.head_scratch[6];

            // 扩展目标缓冲区，预留足够空间存储当前记录片段
            dst.resize(dst_offset + length as usize, 0);
            // 读取记录数据到目标缓冲区
            bytes_read = self.src.read(&mut dst[dst_offset..dst_offset + length as usize])?;
            self.blk_off += bytes_read;

            // 若启用校验和（self.checksums = true），则验证读取的数据是否与头部的 checksum 匹配，不匹配则返回损坏错误。
            if self.checksums && !self.check_integrity(typ, &dst[dst_offset..dst_offset + bytes_read], checksum) {
                return err(StatusCode::Corruption, "Invalid Checksum");
            }

            dst_offset += length as usize;
            // Full（完整记录）：直接返回总长度（dst_offset），读取完成。
            // First（跨块记录的第一部分） 或 Middle（中间部分）：继续循环读取下一部分。
            // Last（跨块记录的最后一部分）：返回总长度，读取完成（此时 dst 已拼接完整记录）。
            if typ == RecordType::Full as u8 {
                return Ok(dst_offset);
            } else if typ == RecordType::First as u8 {
                continue;
            } else if typ == RecordType::Middle as u8 {
                continue;
            } else if typ == RecordType::Last as u8 {
                return Ok(dst_offset);
            }
        }
    }

    fn check_integrity(&mut self, typ: u8, data: &[u8], expected: u32) -> bool {
        let mut digest = crc::digest();
        digest.update(&[typ]);
        digest.update(data);
        unmask_crc(expected) == digest.finalize()
    }
}

const MASK_DELTA: u32 = 0xa282ead8;

/// 这个 mask_crc 函数是 CRC 校验和的 “掩码处理” 工具，用于对原始 CRC 校验和进行转换处理后再存储到磁盘。这是 LevelDB/RocksDB 等数据库中为提升校验和可靠性而设计的优化手段，与 unmask_crc 函数配合使用（mask_crc 用于存储前的掩码，unmask_crc 用于读取时的解掩码）。
///
/// 核心背景：为什么需要掩码 CRC？
///
/// 原始 CRC 校验和在某些特殊场景下可能存在 “误判风险”：
///
/// 例如全零数据块的 CRC 校验和也是零，可能与未初始化的内存区域（默认值为零）混淆；
///
/// 某些数据模式可能恰好生成与错误数据相同的校验和，导致漏检。
///
/// 通过掩码处理，可以打破这种 “特殊值关联”，使校验和分布更随机，降低误判概率。
///
/// 核心逻辑解析
///
/// 函数接收原始 CRC 校验和（c: u32），返回掩码处理后的 CRC 值（u32），处理步骤如下：
pub fn mask_crc(c: u32) -> u32 {
    // 对原始 CRC 值执行循环移位：
    // wrapping_shr(15)：无符号右移 15 位（高位补 0）；
    // wrapping_shl(17)：无符号左移 17 位（低位补 0）；
    // |（按位或）将两个结果合并，实现 “循环右移 15 位” 的效果（等价于循环左移 32-15=17 位）。
    // 移位的目的是打乱原始 CRC 的位分布，避免特殊值（如全零）的校验和仍为特殊值。
    // MASK_DELTA 是一个预定义常量（通常为 0xa282ead8），用于进一步偏移校验和的值；
    // wrapping_add 确保加法溢出时按 u32 类型正确环绕（避免溢出错误）。
    (c.wrapping_shr(15) | c.wrapping_shl(17)).wrapping_add(MASK_DELTA)
}

/// 这个 unmask_crc 函数是CRC 校验和的 “解掩码” 工具，用于将存储在磁盘上的 “掩码 CRC 值” 还原为原始的 CRC 校验和，以便与计算出的校验和进行比对。
/// 这是 LevelDB/RocksDB 等 LSM 树数据库中常见的校验和存储优化手段。
///
/// 核心背景：CRC 掩码的作用
///
/// 为了避免某些特殊场景下的误判（例如全零数据块的校验和也是零，可能与未初始化内存混淆），数据库在存储 CRC 校验和时会先进行 “掩码处理”（mask），将原始 CRC 值转换为一个不同的值。而 unmask_crc 则是掩码处理的逆操作，用于读取时恢复原始 CRC 值。
///
/// 核心逻辑解析
///
/// 函数接收一个掩码后的 CRC 值（mc: u32），返回原始的 CRC 校验和（u32），处理步骤如下：
pub fn unmask_crc(mc: u32) -> u32 {
    // MASK_DELTA 是一个预定义的常量（通常为 0xa282ead8），用于第一步偏移计算；
    // wrapping_sub 是 Rust 中的 “环绕减法”，确保即使结果为负数也会按 u32 类型正确环绕（避免溢出错误）。
    let rot = mc.wrapping_sub(MASK_DELTA);
    // 对 rot 执行循环右移 17 位和循环左移 15 位的组合操作：
    // wrapping_shr(17)：将 rot 无符号右移 17 位，高位补 0；
    // wrapping_shl(15)：将 rot 无符号左移 15 位，低位补 0；
    // |（按位或）将两个移位结果合并，实现循环移位效果（右移 17 位等价于左移 32-17=15 位的循环移位）。
    rot.wrapping_shr(17) | rot.wrapping_shl(15)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Cursor;

    #[test]
    fn test_crc_mask_crc() {
        let mut digest = crc::digest();
        digest.update("abcde".as_bytes());
        let sum = digest.finalize();
        assert_eq!(sum, unmask_crc(mask_crc(sum)));
        assert!(sum != mask_crc(sum));
    }

    #[test]
    fn test_crc_sanity() {
        assert_eq!(0x8a9136aa, crc::crc32([0_u8; 32]));
        assert_eq!(0x62a8ab43, crc::crc32([0xff_u8; 32]));
    }

    #[test]
    fn test_writer() {
        let data = &[
            "hello world. My first log entry.",
            "and my second",
            "and my third",
        ];
        let mut lw = LogWriter::new(Vec::new());
        let total_len = data.iter().fold(0, |l, d| l + d.len());
        println!("{}", total_len);

        for d in data {
            let _ = lw.add_record(d.as_bytes());
        }

        assert_eq!(lw.current_block_offset, total_len + 3 * super::HEADER_SIZE);
    }

    #[test]
    fn test_writer_append() {
        let data = &[
            "hello world. My first log entry.",
            "and my second",
            "and my third",
        ];

        let mut dst = vec![0_u8; 1024];

        {
            let mut lw = LogWriter::new(Cursor::new(dst.as_mut_slice()));
            for d in data {
                let _ = lw.add_record(d.as_bytes());
            }
        }

        let old = dst.clone();

        // Ensure that new_with_off positions the writer correctly. Some ugly mucking about with
        // cursors and stuff is required.
        {
            let offset = data[0].len() + super::HEADER_SIZE;
            let mut lw =
                LogWriter::new_with_off(Cursor::new(&mut dst.as_mut_slice()[offset..]), offset);
            for d in &data[1..] {
                let _ = lw.add_record(d.as_bytes());
            }
        }
        assert_eq!(old, dst);
    }

    #[test]
    fn test_reader() {
        let data = [
            "abcdefghi".as_bytes().to_vec(),    // fits one block of 17
            "123456789012".as_bytes().to_vec(), // spans two blocks of 17
            "0101010101010101010101".as_bytes().to_vec(),
        ]; // spans three blocks of 17
        let mut lw = LogWriter::new(Vec::new());
        lw.block_size = super::HEADER_SIZE + 10;

        for e in data.iter() {
            assert!(lw.add_record(e).is_ok());
        }

        assert_eq!(lw.dst.len(), 93);
        // Corrupt first record.
        lw.dst[2] += 1;

        let mut lr = LogReader::new(lw.dst.as_slice(), true);
        lr.blocksize = super::HEADER_SIZE + 10;
        let mut dst = Vec::with_capacity(128);

        // First record is corrupted.
        assert_eq!(
            err(StatusCode::Corruption, "Invalid Checksum"),
            lr.read(&mut dst)
        );

        let mut i = 1;
        loop {
            let r = lr.read(&mut dst);

            if r.is_err() {
                panic!("{}", r.unwrap_err());
            } else if r.unwrap() == 0 {
                break;
            }

            assert_eq!(dst, data[i]);
            i += 1;
        }
        assert_eq!(i, data.len());
    }

    #[test]
    fn test_mask_crc() {
        let mut digest = crc::digest();
        digest.update("abcde".as_bytes());
        let sum = digest.finalize();
        println!("{}", sum);
        let i = mask_crc(sum);
        println!("{}", i);
        let j = unmask_crc(i);
        println!("{}", j);
    }
}
